package com.zang.test.rocketmq.service;

import com.zang.test.rocketmq.constant.MQConstant;
import com.zang.test.rocketmq.model.MsgEvt;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 发送服务
 * @author Zhang Qiang
 * @date 2019/10/31 15:38
 */
@Slf4j
@Service
public class SendMqMessageService {

    @Resource
    private RocketMQTemplate rocketMqTemplate;

    private final String q_topic = MQConstant.QUEUE_RESPONSE_TOPIC;

    public String sendMessage(String str) {
        AtomicInteger atomicInteger = new AtomicInteger();
        final String var1 = new String(str.getBytes(), StandardCharsets.UTF_8);
        Message message = new Message(MQConstant.QUEUE_MESSAGE_KEY, var1.getBytes());
        MsgEvt msgEvt = MsgEvt.builder().orderId("1").message(message).build();
        return this.syncSend(q_topic, msgEvt);
    }

    /**
     * 异步推送消息队列
     * @param destination 目的topic
     * @param msgEvt 消息体
     * @return 消息
     */
    private String asyncSend(String destination, MsgEvt msgEvt){
        String var1 = new String(msgEvt.getMessage().getBody(), StandardCharsets.UTF_8);
        rocketMqTemplate.asyncSend(destination, msgEvt, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info(" ===== onSuccess ===== 消息已发送： {} ", var1);
            }
            @Override
            public void onException(Throwable throwable) {
                log.info(" ===== onException ===== : {} , throwable : {}", var1, throwable.getMessage());
            }
        });
        return var1;
    }

    private String syncSend(String destination, MsgEvt evt){
        String var1 = new String(evt.getMessage().getBody(), StandardCharsets.UTF_8);
        rocketMqTemplate.syncSend(destination, evt);
        log.info(" 发送消息 {} ", var1);
        return var1;
    }


}
